# -*- coding: utf-8 -*-
# import gevent.monkey
# gevent.monkey.patch_all()
import gevent
import sys
import time
import random
import numpy as np
import pandas as pd

from contract_market.utils.utils import ToolsMixinsStructer
from contract_market.utils.decorator import loop, exception
from .algorithm import QuotoAlgorithm
from .type import Action
from contract_market.utils.api import xt_api, binance_api
from contract_market.strategy.conf import DynamicAllocation
from contract_market.strategy.image import constant
from contract_market.strategy.image.pricing import Pricing
from contract_market.strategy.image.utils import send_worktile_msg
from contract_market.strategy.image.runtime_params import wait_params, optimal_status, redis_optimal_state
from contract_market.strategy.image.optimal import Optimal
from contract_market.utils.redisdb import redisdb


class OrderBookUtils(QuotoAlgorithm):

    def __init__(self, configure) -> None:
        super().__init__()
        self.symbol = configure.argv.symbol_name
        self.cfg = DynamicAllocation(configure, constant)
        print('da cfg', self.cfg.values())
        self.optimal = Optimal(self.cfg, self.symbol, self.cfg.values('transcation_type'))
        self.pricing = Pricing()
        self.verbose = True

    # utils
    def polynomial(self, y: list):
        x = [i for i in range(len(y))]
        c = np.polyfit(x, y, deg=self.cfg.values('degress'))
        hat_y = [np.sum([c[j]*i**(self.cfg.values('degress')-j) for j in range(self.cfg.values('degress')+1)]) for i in x]
        return c, hat_y

    def differential(self, c1, c2, inc=True):
        if inc:
            c = (np.array(c1)-np.array(c2))/np.array(c2)
        return c

    def balance_to_csize(self, balance):
        if self.cfg.values('transcation_type') == 'um':
            # 单位是张
            # um: 1张=contractSize个base asset
            # cm: 1张=contractSize个auote asset
            csize = balance / xt_api.index_price(self.symbol) / xt_api.contract_size
        elif self.cfg.values('transcation_type') == 'cm':
            csize = balance * xt_api.index_price(self.symbol) / xt_api.contract_size
        else:
            raise NotImplementedError
        return csize

    def quantity_to_csize(self, quantity):
        if self.cfg.values('transcation_type') == 'um':
            csize = quantity / xt_api.contract_size
        elif self.cfg.values('transcation_type') == 'cm':
            csize = quantity / xt_api.contract_size
        return csize

    def obsevation(self, obs):
        '''
        support user or mt(mid+tail)
            please change obs to provide support(user+mid+tail)
        '''
        obsevation = {
            'user': (0, self.cfg.values('user_obs')),
            'mid': (0, self.cfg.values('mid_obs')),
            'tail': (self.cfg.values('mid_obs'), self.cfg.values('max_order'))
        }
        if obs == 'user':
            obsevation.pop('mid')
            obsevation.pop('tail')
        elif obs == 'mt':
            obsevation.pop('user')
        return obsevation
    
    def pricing_data(self, s, interval=5):
        if int(time.time()) % interval:
            return 0, 0, (0, 0)
        # position
        pos = xt_api.position(self.symbol)
        pos = pd.DataFrame(pos)
        lp = float(pos[pos.positionSide=='LONG'].availableCloseSize.squeeze())
        sp = float(pos[pos.positionSide=='SHORT'].availableCloseSize.squeeze())
        # real market vol
        # binance kline, unit second
        kline = binance_api.kline(self.cfg.values('mirror_market'))
        sigma = (np.log(kline.close_price/kline.open_price)).std()
        # pricing
        params = {
            'lp': lp,
            'sp': sp,
            's': s,
            'sigma': sigma,
        }
        opt_price, opt_spread, opt_l1 = self.pricing.as_model(**params)
        return opt_price, opt_spread, opt_l1

    # core
    def mirror_data(self, observation):
        t0 = time.time()
        mirror = self.construct_price(observation)
        t1 = time.time()
        mirror_quantity = self.construct_quantity(observation)
        t2 = time.time()
        mirror['quantity'] = mirror_quantity
        mirror = mirror[(mirror.price>0) & (mirror.quantity>0)]
        t3 = time.time()
        if self.verbose:
            num = self.cfg.values('user_obs') if observation == 'user' else \
                self.cfg.values('max_order')-self.cfg.values('user_obs')
            print('mirror loss:', observation, num*2, len(mirror), num*2-len(mirror))
            print('mirror data(l-s):', mirror[mirror.side==Action.long].price.min(), \
                mirror[mirror.side==Action.long].price.max(), \
                mirror[mirror.side==Action.short].price.min(), \
                mirror[mirror.side==Action.short].price.max())
            print('mirror_data timing', t1-t0, t2-t1, t3-t2)
        return mirror

    def gen_price(self):
        user_long_range, user_short_range = self.cfg.values('user_long_range'), self.cfg.values('user_short_range')
        _mark_price = xt_api.index_price(self.symbol)
        _index_price = xt_api.index_price(self.symbol, type='index_price')
        # insert index/mark price
        long_price, short_price = min(_mark_price, _index_price), max(_mark_price, _index_price)
        # add as model - price
        opt_price, opt_spread, opt_l1 = self.pricing_data(_index_price)
        if self.cfg.values('use_as_pricing') and opt_price:
            c = opt_price / _index_price
            long_price *= c
            short_price *= c
        # add as model - spread
        if self.cfg.values('use_as_spread') and opt_spread:
            long_price, short_price = long_price-(long_price*opt_spread*0.5), short_price+(short_price*opt_spread*0.5)
        # add default spread
        long_price *= (1-self.cfg.values('spread'))
        short_price *= (1+self.cfg.values('spread'))
        # add optimal
        if not int(time.time()) % wait_params.get('optimal_time'):
            c1, c2, c3, c4 = 0, 0, (0,0), 0
            if self.cfg.values('balance_density_optimal'):
                c1 = self.optimal.balance_density()
                user_long_range += c1
                user_short_range += c1
            if self.cfg.values('balance_spread_optimal'):
                c2 = self.optimal.balance_spread()
                long_price *= (1-c2)
                short_price *= (1+c2)
            if self.cfg.values('position_density_optimal'):
                c3 = self.optimal.position_density()
                user_long_range += c3[0]
                user_short_range += c3[1]
            if self.cfg.values('position_spread_optimal'):
                c4, side = self.optimal.position_spread()
                if side == 'LONG':
                    long_price *= (1-c4)
                else:
                    short_price *= (1+c4)
            # 
            content = f'{self.symbol} optimal c1/c3/c4/c5/c6: {c1, c2, c3, c4}'
            send_worktile_msg(content)
            print(content)
        return long_price, short_price, user_long_range, user_short_range

    def construct_price(self, observation):
        '''
        user_obs + mid + com
        return
            price: pd.DataFrame([...])
        '''
        # load conf
        user_obs = self.cfg.values('user_obs')
        # common
        c = [1] * (self.cfg.values('degress')+1)
        def poly(i):
            return [c[j]*i**(self.cfg.values('degress')-j) for j in range(self.cfg.values('degress')+1)]
        t0 = time.time()
        # pricing
        long_price, short_price, user_long_range, user_short_range = self.gen_price()
        print('gen_price', long_price, short_price, user_long_range, user_short_range)
        # generate user/mid/tail obs
        t1 = time.time()
        user_short, user_long, mid_short, mid_long, tail_short, tail_long = [], [], [], [], [], []
        if observation == 'user':
            if self.cfg.values('is_mp'):
                for i in range(1, user_obs+1):
                    user_long.append(long_price-i*xt_api.price_step)
                    user_short.append(short_price+i*xt_api.price_step)
            else:
                for i in np.linspace(0, user_long_range, user_obs+1):
                    if i == 0: continue
                    user_long.append((2-np.sum(poly(i)))*long_price)
                for i in np.linspace(0, user_short_range, user_obs+1):
                    if i == 0: continue
                    user_short.append(np.sum(poly(i))*short_price)
        elif observation == 'mt':
            # pos user
            if self.cfg.values('is_mp'):
                long_lastprice = long_price-user_obs*xt_api.price_step
                short_lastprice = short_price+user_obs*xt_api.price_step
            else:
                long_lastprice = (2-np.sum(poly(user_long_range)))*long_price
                short_lastprice = np.sum(poly(user_short_range))*short_price
            for i in np.linspace(0, self.cfg.values('mid_range'), self.cfg.values('mid_obs')):
                mid_long.append((2-np.sum(poly(i)))*long_lastprice)
                mid_short.append(np.sum(poly(i))*short_lastprice)
            for i in np.linspace(0, self.cfg.values('tail_range'), self.cfg.values('tail_obs')):
                tail_long.append((2-np.sum(poly(i)))*mid_long[-1])
                tail_short.append(np.sum(poly(i))*mid_short[-1])
        else:
            raise NotImplementedError
            
        # coefs
        short_price = np.array(user_short + mid_short + tail_short) * self.cfg.values('custom_short_price')
        long_price = np.array(user_long + mid_long + tail_long) * self.cfg.values('custom_long_price')
        # convert
        short = pd.DataFrame(short_price, columns=['price'])
        short['side'] = Action.short
        long = pd.DataFrame(long_price, columns=['price'])
        long['side'] = Action.long
        mirror = pd.concat([long, short])
        #
        mirror['price'] = mirror.price.apply(lambda x: xt_api.precision(x, 'price'))
        # mirror['quantity'] = mirror.price.apply(lambda x: xt_api.precision(x, 'quantity'))
        t2 = time.time()
        if self.verbose:
            print('construct_price timing', t1-t0, t2-t1)
        return mirror

    def construct_quantity(self, observation):
        '''
        coef*mean
        return
            quantity: pd.DataFrame([...])
        '''
        # load config 
        mu = self.cfg.values('mu')
        sigma = self.cfg.values('sigma')
        com_range = self.cfg.values('com_range')
        max_order = self.cfg.values('max_order')
        degress = self.cfg.values('degress')
        user_obs = self.cfg.values('user_obs')
        
        #
        c = [1] * (degress+1)
        linspace = np.linspace(0, com_range, max_order)
        
        #
        if observation == 'user':
            fund_ratio = 0.5*user_obs/max_order
            c1_range = user_obs
            linspace = linspace[:user_obs]
        elif observation == 'mt':
            fund_ratio = 0.5*(max_order-user_obs)/max_order
            c1_range = max_order-user_obs
            linspace = linspace[user_obs:]
        k = self.cfg.values('k')
        
        #
        quantity = [np.sum([c[j]*i**(degress-j) for j in range(degress+1)])*(1+random.gauss(mu, sigma)) \
            for i in linspace]
        
        # random process of top n
        r1 = random.sample([i for i in range(int(user_obs*0.3))], k=k)
        c1 = [random.randint(1,20)*0.05 if i in r1 else 1 for i in range(c1_range)]
        quantity *= np.array(c1)
        
        #
        unit = self.symbol.split('_')[1] if self.cfg.values('transcation_type') == 'um' else self.symbol.split('_')[0]
        wallet_balance = xt_api.balance(method='api', unit=unit, kind='walletBalance')
        balance = self.cfg.values('max_fund')*(1-random.randint(1,200)/1000) \
            if wallet_balance > self.cfg.values('max_fund') else wallet_balance
        # add liquid control
        loss = self.cfg.values('init_fund')-wallet_balance
        loss = loss if loss > 0 else 0
        balance = balance if self.cfg.values('keep_liquid_control') else balance-loss
        csize = self.balance_to_csize(balance)
        quantity = np.array(quantity) * (csize*fund_ratio*self.cfg.values('capital_rate')/np.sum(quantity))
        
        #
        short_quantity = quantity * (1-random.randint(1,100)/1000) * \
            self.cfg.values('custom_short_quantity')
        short_quantity = pd.DataFrame(short_quantity, columns=['quantity'])
        long_quantity = quantity * (1-random.randint(1,100)/1000) * \
            self.cfg.values('custom_long_quantity')
        long_quantity = pd.DataFrame(long_quantity, columns=['quantity'])
        mirror_quantity = pd.concat([long_quantity, short_quantity])
        return mirror_quantity

    def patch_action(self, cancel_action, send_action, maker_obs):
        # patch cancel
        ck = self.cfg.values('k') if len(cancel_action)>self.cfg.values('k') else len(cancel_action)      # k-filter 
        ck = 0 if len(maker_obs) < int(self.cfg.values('max_order')*0.8) else ck     # init-filter 
        ck = self.cfg.values('singel_order') if ck > self.cfg.values('singel_order') else ck       # singel-filter
        # patch send
        sk = self.cfg.values('k') if len(send_action)>self.cfg.values('k') else len(send_action)      # k-filter 
        sk = len(send_action) if len(maker_obs) < int(self.cfg.values('max_order')*0.8) else sk     # init-filter
        sk = self.cfg.values('singel_order') if sk > self.cfg.values('singel_order') else sk       # singel-filter
        return ck, sk

    # overflow
    def find_overflow(self, maker, side):
        '''
        extream price overflow
        order number overflow
        '''
        # load conf
        max_order = self.cfg.values('max_order')
        print('max_order:', max_order)
        #
        if maker.empty:
            return [], []
        ascending = False if side == 'long' else True
        orders = maker[maker.side==Action.long] if side == 'long' else maker[maker.side==Action.short]
        orders = orders.sort_values('price', ascending=ascending)
        if len(orders) < max_order:
            return [], []

        # outside
        outside = orders[int(max_order*0.8):]
        n = self.cfg.values('singel_order') if len(outside)>self.cfg.values('singel_order') else len(outside)
        outside = outside.sample(n=n)
        
        o_action = pd.DataFrame() if outside.empty else outside
        o_action = o_action.to_dict('records')
        print('find_overflow', side, len(o_action))
        return o_action

    # intersection
    def find_outside_range(self, maker, side, is_is=False):
        long_maxp = short_minp = xt_api.index_price(self.symbol)
        # long_minp, long_maxp = mirror[mirror.side==Action.long].price.min(), mirror[mirror.side==Action.long].price.max()
        # short_minp, short_maxp = mirror[mirror.side==Action.short].price.min(), mirror[mirror.side==Action.short].price.max()
        maker_long, maker_short = maker[maker.side==Action.long], maker[maker.side==Action.short]
        if side == 'long':
            if is_is:
                action = pd.DataFrame() if maker_long.empty else maker_long[maker_long.price>=long_maxp]
            else:
                action = pd.DataFrame() # if maker_long.empty else maker_long[maker_long.price<=long_minp]
        elif side == 'short':
            if is_is:
                action = pd.DataFrame() if maker_short.empty else maker_short[maker_short.price<=short_minp]
            else:
                action = pd.DataFrame() # if maker_short.empty else maker_short[maker_short.price>=short_maxp]
        else:
            raise NotImplementedError
        if not action.empty and self.verbose:
            print('find_outside_range', len(action), action.price.to_list())
        action = action.to_dict('records')
        return action

    def exec_outside_action(self, action, method='asyn'):
        # exec action
        if method == 'single':
            for act in action:
                xt_api.cancel(symbol=self.symbol, action=act, method='sf')
        elif method == 'asyn':
            jobs = [gevent.spawn(xt_api.cancel, **{'symbol': self.symbol, 'action': act}) for act in action]
            gevent.joinall(jobs, timeout=wait_params.get('timeout'))
        elif method == 'batch':
            xt_api.cancel(symbol=self.symbol, action=action)
        return True

    # position
    def find_sync_action(self, mirror, maker, side, observation):
        obs = self.obsevation(obs=observation)
        user_action, mid_action, tail_action = [], [], []
        for key, value in obs.items():
            # find price range
            mirror_long_obs, mirror_short_obs = mirror[mirror.side==Action.long].iloc[value[0]:value[1]], \
                mirror[mirror.side==Action.short].iloc[value[0]:value[1]]
            mirror_long_min, mirror_long_max = mirror_long_obs.price.min(), mirror_long_obs.price.max()
            mirror_short_min, mirror_short_max = mirror_short_obs.price.min(), mirror_short_obs.price.max()
            maker_long, maker_short = maker[maker.side==Action.long], maker[maker.side==Action.short]
            maker_long_obs, maker_short_obs = maker_long[(maker_long.price>=mirror_long_min)&(maker_long.price<=mirror_long_max)], \
                maker_short[(maker_short.price>=mirror_short_min)&(maker_short.price<=mirror_short_max)]
            # find cancel/send price
            if side == 'long':
                maker_obs, mirror_obs = maker_long_obs, mirror_long_obs 
            elif side == 'short':
                maker_obs, mirror_obs = maker_short_obs, mirror_short_obs
            else:
                return False
            # construct action
            cancel_action, send_action = [], []
            cancel = list(set(maker_obs.price.to_list()).difference(set(mirror_obs.price.to_list())))
            for c in cancel:
                for mc in maker_obs[maker_obs.price == c].to_dict('records'):
                    cancel_action.append({'type': 'cancel', 'price': mc.get('price'), \
                        'quantity': mc.get('quantity'), 'side': mc.get('side'), 'id': mc.get('id')})
            send = list(set(mirror_obs.price.to_list()).difference(set(maker_obs.price.to_list())))
            for s in send:
                c = mirror_obs[mirror_obs.price==s].iloc[0]
                send_action.append({'type': 'send', 'price': c.price, 'quantity': c.quantity, 'side': c.side, 'id': 0})

            # classify
            if key == 'user':     # stream
                print('sync user number', len(cancel), len(send))
                send_action = random.sample(send_action, len(send_action)//2)
                user_action = cancel_action + send_action
            elif key == 'mid':    # asyn+sample
                # patch k
                ck, sk = self.patch_action(cancel_action, send_action, maker_obs)
                # agg
                cancel_action = random.sample(cancel_action, k=ck)
                send_action = random.sample(send_action, k=sk)
                mid_action = cancel_action + send_action
                print('sync mid number', len(cancel), cancel, '\n', len(send), send)
            elif key == 'tail':   # asyn+sample
                # patch k
                ck, sk = self.patch_action(cancel_action, send_action, maker_obs)
                # agg
                cancel_action = random.sample(cancel_action, k=ck)
                send_action = random.sample(send_action, k=sk)
                tail_action = cancel_action + send_action
                print('sync tail number', len(cancel), cancel, '\n', len(send), send)
        # print(f'mt price: long={mirror[mirror.side==Action.long].price.max()}, short={mirror[mirror.side==Action.short].price.min()}')
        return user_action, mid_action, tail_action

    def exec_sync_action(self, user_action, mid_action, tail_action, side='short', method='asyn'):
        reverse = False if side == 'short' else True
        # optimal action
        if user_action:
            user_action = sorted(user_action, key=lambda k: (k['type'], k['price']), reverse=reverse)
        if mid_action:
            mid_action = sorted(mid_action, key=lambda k: (k['type'], k['price']), reverse=reverse)
        if tail_action:
            tail_action = sorted(tail_action, key=lambda k: (k['type'], k['price']), reverse=reverse)
        # asyn exec
        def job_user():
            for act in user_action:
                xt_api.send(symbol=self.symbol, action=act) if act.get('type')=='send' \
                    else xt_api.cancel(symbol=self.symbol, action=act)
                gevent.sleep(0.1)
        def job_mid():
            for act in mid_action:
                xt_api.send(symbol=self.symbol, action=act) if act.get('type')=='send' \
                    else xt_api.cancel(symbol=self.symbol, action=act)
                gevent.sleep(0.1)
        def job_tail():
            for act in tail_action:
                xt_api.send(symbol=self.symbol, action=act) if act.get('type')=='send' \
                    else xt_api.cancel(symbol=self.symbol, action=act)
                gevent.sleep(0.1)
        #
        if method == 'single':
            job_user(); job_mid(); job_tail()
        elif method == 'asyn':
            jobs = [gevent.spawn(job) for job in [job_user, job_mid, job_tail]]
            gevent.joinall(jobs, timeout=wait_params.get('timeout'))
            if self.verbose:
                print('exec_sync_action number(u/m/t)', len(user_action), len(mid_action), len(tail_action))
        return True

    def price_sync(self, maker, mirror, side, observation):
        '''
        params
            maker: DataFrame(columns=['price', 'quantity', 'side', 'id'])
            mirror: DataFrame([columns=['price', 'quantity', 'side'])
        '''
        is_is = True if observation == 'user' else False
        # init orderbook
        t0 = time.time()
        if maker.empty:
            self.init_orderbook(mirror)
            return True
        
        # cancel of outside the price range
        t1 = time.time()
        tmp_side = 'short' if side == 'long' else 'long'
        action = self.find_outside_range(maker, tmp_side, is_is=is_is)
        self.exec_outside_action(action)
        t2 = time.time()

        # sync price range
        user_action, mid_action, tail_action = self.find_sync_action(mirror, maker, side, observation=observation)
        self.exec_sync_action(user_action, mid_action, tail_action, side=side)
        t3 = time.time()
        if self.verbose:
            print('price_sync number(u/m/t)', len(user_action), len(mid_action), len(tail_action))
            print('price_sync timing(i/c/s)', t1-t0, t2-t1, t3-t2)
        return True


class OrderBookMixins(ToolsMixinsStructer, OrderBookUtils):

    def __init__(self, configure, event, engine) -> None:
        super().__init__(configure)
        self.event = event
        self.engine = engine

    def init_xtapi(self, event, engine, verbose=False):
        try:
            xt_api.init_engine(event, engine)
            cli_res = xt_api.init_client(self.symbol, self.cfg.values('transcation_type'))
            conf_res = xt_api.init_config(self.symbol)
            acc_res = xt_api.init_account(self.symbol)
            if verbose:
                print(cli_res, conf_res, acc_res)
        except Exception as ex:
            print(ex)
            print('Configuration initialization failed. Try again ...')
            time.sleep(3)
            self.init_xtapi(event, engine)
        if xt_api.price_prec is None or xt_api.price_step is None or \
            xt_api.quantity_prec is None or xt_api.contract_size is None:
            raise BaseException('Error loading configuration from XT API')
        return True

    def init_optimal_state(self):
        print('Init optimal state')
        for key, value in optimal_status.items():
            redisdb.standalone_operator('hset', self.symbol+redis_optimal_state, (key, str(value)))
        return True

    def init_orderbook(self, mirror):
        mirror['type'] = 'send'
        mirror['id'] = 0
        actions = mirror.to_dict('records')
        for action in actions:
            time.sleep(0.5)
            xt_api.send(self.symbol, action)
        if self.verbose:
            print('init orderbook', len(actions))
        return True

    def stop_all(self):
        res = xt_api.xt.close_all_position()
        print('stop all: close position', res)
        res = xt_api.xt.cancel_all_order()
        print('stop all: cancel order', res)
        return True

    def timer(self):
        value = redisdb.standalone_operator('hget', self.symbol+redis_optimal_state, 'temp_stop_liquid')
        if int(value) <= 0:
            return True
        value = int(value) - 1
        redisdb.standalone_operator('hset', self.symbol+redis_optimal_state, ('temp_stop_liquid', value))
        print(f'update temp_stop_send_order state: {value+1} to {value}')
        time.sleep(1)
        return True

    # user price sync
    def user_long_price_sync(self, maker, mirror):
        self.price_sync(maker, mirror, 'long', observation='user')
        return True
    
    def user_short_price_sync(self, maker, mirror):
        self.price_sync(maker, mirror, 'short', observation='user')
        return True
    
    # mt price sync
    def mt_long_price_sync(self, maker, mirror):
        self.price_sync(maker, mirror, 'long', observation='mt')
        return True
    
    def mt_short_price_sync(self, maker, mirror):
        self.price_sync(maker, mirror, 'short', observation='mt')
        return True

    # quantity
    def quantity_sync(self, maker, mirror):
        '''
        len(maker) == len(mirror)
        return: DataFrame
            quantity action: [quantity]
        '''
        maker_obs, mirror_obs = maker, mirror
        maker_obs, mirror_obs = maker.iloc[self.cfg.values('user_obs'):self.cfg.values('user_obs')+self.cfg.values('mid_obs')], \
            mirror.iloc[:self.cfg.values('mid_obs')]
        if len(maker_obs)!=len(mirror_obs):
            print('quantity_sync failed', len(maker_obs), len(mirror_obs))
            return False
        #
        _, maker_haty = self.polynomial(maker_obs.quantity)
        _, mirror_haty = self.polynomial(mirror_obs.quantity)
        #
        dc = self.differential(mirror_haty, maker_haty)
        quantity = np.array(maker_obs.quantity) * np.array(dc)
        maker_obs2 = maker_obs.copy()
        maker_obs2.loc[:,'quantity'] = quantity
        #
        top = maker_obs2[maker_obs2.quantity > 0]
        if len(top) > self.cfg.values('k'):
            top = top.sample(self.cfg.values('k'))
        for act in top.to_dict('records'):
            xt_api.send(symbol=self.symbol, action=act)
        print('quantity_sync', top)
        return True
    
    # trade
    def trade_sync_double(self):
        ts = int(time.time())%60
        if all([20<ts<40, random.random()>0.5]):
            return False
        # trade action
        price = xt_api.index_price(self.symbol)
        if random.random() > 0.3:
            # TODO: fitting xt config: ref xx_to_csize
            if self.cfg.values('transcation_type') == 'um':
                qty = random.randint(10, 100)
            else:
                qty = random.randint(100, 500)
            act_long = {'type': 'send', 'price': price, 'quantity': qty, 'side': Action.long, 'id': 0}
            act_short = {'type': 'send', 'price': price, 'quantity': qty, 'side': Action.short, 'id': 0}
        else:
            action = binance_api.trades(symbol=self.cfg.values('mirror_market'))
            if not action:
                return False
            if self.cfg.values('transcation_type') == 'um':
                qty = random.randint(10, 100) + action['quantity']*self.cfg.values('trade_scale')
            else:
                qty = random.randint(100, 500) + action['quantity']*self.cfg.values('trade_scale')
            act_long = action.copy()
            act_long['side'] = Action.long
            act_long['quantity'] = qty
            act_long['price'] = price
            act_short = action.copy()
            act_short['side'] = Action.short
            act_short['quantity'] = qty
            act_short['price'] = price
            if not all([act_long, act_short]):
                return False

        act_long['quantity'] = self.quantity_to_csize(act_long['quantity'])
        act_short['quantity'] = self.quantity_to_csize(act_short['quantity'])
        # TODO: check abnormal
        res1 = xt_api.send(symbol=self.symbol, action=act_long, timeInForce='GTC', mode='open')
        if isinstance(res1, bool):
            return False
        id = res1.get('result')
        res2 = xt_api.send(symbol=self.symbol, action=act_short, timeInForce='IOC', mode='open')
        cancel_long = {'type': 'cancel', 'price': 0, 'quantity': 0, 'side': 0, 'id': id}
        res3 = xt_api.cancel(symbol=self.symbol, action=cancel_long, method='api')
        if self.verbose:
            print(f'trade_sync send {act_long, act_short, res1, res2}')
        return True

    # density
    def sub_density(self, maker, mirror, side, observation='user'):
        '''
        cumsum orders
        '''
        ascending = False if side == 'long' else True
        ss = Action.long if side == 'long' else Action.short
        maker_ls = maker[maker.side==ss]
        if len(maker_ls) < self.cfg.values('max_order')*0.8:
            return False
        obs = self.obsevation(obs=observation)

        for key, value in obs.items():
            # find price range
            mirror_long = mirror[mirror.side==ss].iloc[value[0]:value[1]]
            mirror_long_min, mirror_long_max = mirror_long.price.min(), mirror_long.price.max()
            order = maker_ls[(maker_ls.price>=mirror_long_min)&(maker_ls.price<=mirror_long_max)]
            # user density processing
            if key=='user':
                if self.cfg.values('is_mp'):
                    density = 0
                else:
                    density = self.cfg.values(key+'_long_range') if side == 'long' \
                        else self.cfg.values(key+'_short_range')
            else:
                density = self.cfg.values(key+'_range')/self.cfg.values(key+'_obs')
            #
            user_order = order.sort_values('price', ascending=ascending)
            cancel_id = user_order[user_order.price.pct_change().abs()<=density].id.to_list()
            if len(user_order) <= self.cfg.values(key+'_obs'):
                return True
            if len(user_order)-len(cancel_id)>=self.cfg.values(key+'_obs'):
                k = int(len(cancel_id) * 0.8)
            else:
                k = int(len(cancel_id) * 0.2)
            cancel_id = sorted(random.sample(cancel_id, k=k))
            action = [{'type': 'cancel', 'price': 0, 'quantity': 0, 'side': 0, 'id': id} for id in cancel_id]
            if self.verbose:
                print(f'sub_density {key, len(mirror_long), len(action)}')
            # batch 
            xt_api.cancel(symbol=self.symbol, action=action)
        return True

    # aux
    def aux_orderbook(self, maker):
        if maker.empty:
            return True
        long = maker[maker.side==Action.long].sort_values('price', ascending=False)
        short = maker[maker.side==Action.short].sort_values('price', ascending=True)
        max_order = self.cfg.values('max_order')

        # s1 - insert price(max diff) -2
        # NOTE: inf inc
        @exception
        def job1():
            long_top, short_top = long[long.price.pct_change()!=0], short[short.price.pct_change()!=0]
            long_top, short_top = long.iloc[5:20], short.iloc[5:20]
            top = long_top
            vol = top.price.pct_change().to_list()[1:]
            if vol and isinstance(vol, list):
                index = vol.index(max(vol))
                s1_send_action = top.iloc[index].to_dict()
                s1_send_action['price'] = (top.iloc[index].price+top.iloc[index+1].price)*0.5
                s1_send_action['quantity'] = top.quantity.mean() * (1+random.random())
                s1_send_action['type'] = 'send'
                res = xt_api.send(self.symbol, s1_send_action)
                if self.verbose:
                    print('activate_orderbook s1_send:', res)
            # insert short
            top = short_top
            vol = top.price.pct_change().to_list()[1:]
            if vol and isinstance(vol, list):
                index = vol.index(max(vol))
                s1_send_action = top.iloc[index].to_dict()
                s1_send_action['price'] = (top.iloc[index].price+top.iloc[index+1].price)*0.5
                s1_send_action['quantity'] = top.quantity.mean() * (1+random.random())
                s1_send_action['type'] = 'send'
                res = xt_api.send(self.symbol, s1_send_action)
                if self.verbose:
                    print('activate_orderbook s1_send:', res)

        # s2 - cancel quantity(median) - 2
        # NOTE: inf sub
        @exception
        def job2():
            gevent.sleep(0.1)
            if len(long) > max_order*0.6:
                long_top, short_top = long.iloc[3:20], short.iloc[3:20]
                top = long_top
                if not top.empty:
                    qty = top.quantity.sort_values().iloc[int(len(top)*0.5)]
                    s2_cancel_action = top[top.quantity==qty].to_dict('records')[0]
                    s2_cancel_action['type'] = 'cancel'
                    res = xt_api.cancel(self.symbol, s2_cancel_action)
                    if self.verbose:
                        print('activate_orderbook s2_cancel:', res)
            if len(long) > max_order*0.6:
                top = short_top
                if not top.empty:
                    qty = top.quantity.sort_values().iloc[int(len(top)*0.5)]
                    s2_cancel_action = top[top.quantity==qty].to_dict('records')[0]
                    s2_cancel_action['type'] = 'cancel'
                    res = xt_api.cancel(self.symbol, s2_cancel_action)
                    if self.verbose:
                        print('activate_orderbook s2_cancel:', res)

        # s3 - random cancel (top n) - 1
        @exception
        def job3():
            gevent.sleep(0.1)
            if len(long)+len(short) > max_order:
                k = 5
                long_top, short_top = long.iloc[:k], short.iloc[:k]
                r1 = random.choice([i for i in range(k)])
                top = long_top if random.random() > 0.5 else short_top
                if not top.empty:
                    s3_action = top.iloc[r1].to_dict()
                    s3_action['type'] = 'cancel'
                    res = xt_api.cancel(self.symbol, s3_action)
                    if self.verbose:
                        print('aux_orderbook s3_cancel:', res)
    
        # s4 - balance long/short - 3
        @exception
        def job4(start=3, end=15, k=1):
            # long_top, short_top = long[long.price.pct_change()!=0], short[short.price.pct_change()!=0]
            long_top, short_top = long, short
            long_top, short_top = long_top.iloc[start:end], short_top.iloc[start:end]
            ratio = long_top.quantity.sum() / short_top.quantity.sum()
            top = short_top if ratio > 1 else long_top
            
            # replace
            top = top.iloc[start:start*2].sample(k)
            top.quantity = top.quantity*ratio/k*0.5*(1+random.random())
            top['type'] = 'send'
            action = top.to_dict('records')
            for act in action:
                xt_api.send(self.symbol, act)
            print('aux_orderbook s4_send', action)
            top['type'] = 'cancel'
            action = top.to_dict('records')
            for act in action:
                xt_api.cancel(self.symbol, act)
            print('aux_orderbook s4_cancel0', action)
            
            # balance
            # ratio = long_top.quantity.max() / short_top.quantity.max()
            if len(long)+len(short) > max_order:
                long_top, short_top = long_top.iloc[:end], short_top.iloc[:end]
                top = long_top if ratio > 1.5 else short_top
                top = top[top.quantity==top.quantity.max()]
                action = top.to_dict('records')
                xt_api.cancel(self.symbol, action[0])
                print('aux_orderbook s4_cancel1', action)
        
        # s5 - quant in long/short - 1
        @exception
        def job5():
            pos = random.randint(2,5)
            long_single, short_single = long[long.price.pct_change()!=0], short[short.price.pct_change()!=0]
            long_top, short_top = long_single.iloc[pos:pos+1], short_single.iloc[pos:pos+1]
            if long_top.quantity.sum() > short_top.quantity.sum():
                top = short_top
                top.quantity *= long_top.quantity.sum()/top.quantity.sum()*(1+random.random())
            else:
                top = long_top
                top.quantity *= short_top.quantity.sum()/top.quantity.sum()*(1+random.random())
            top['type'] = 'send'
            action = top.to_dict('records')
            for act in action:
                xt_api.send(self.symbol, act)
            print('aux_orderbook s5_send', action)

        # s6 - replace quantity - 5
        @exception
        def job6():
            if len(long)+len(short) < max_order:
                return True
            observation = 'user' if random.random()>0.8 else 'mt'
            if observation == 'user':
                long_density = self.cfg.values('user_long_range') 
                short_density = self.cfg.values('user_short_range') 
            else:
                long_density = short_density = self.cfg.values('mid_range')
            long_top, short_top = long[long.quantity.pct_change()<long_density], short[short.price.pct_change()<short_density]
            lk = self.cfg.values('k') if len(long_top) > self.cfg.values('k') else len(long_top)
            sk = self.cfg.values('k') if len(short_top) > self.cfg.values('k') else len(short_top)
            long_action, short_action = long_top.sample(lk), short_top.sample(sk)
            def job1():
                for action in long_action.to_dict('records'):
                    xt_api.cancel(self.symbol, action)
            def job2():
                for action in short_action.to_dict('records'):
                    xt_api.cancel(self.symbol, action)
            jobs = [gevent.spawn(job) for job in [job1, job2]]
            gevent.joinall(jobs, timeout=wait_params.get('timeout'))
            print('s6_cancel', len(long_action), len(short_action))

        jobs = [gevent.spawn(job) for job in [job1, job2, job3, job4, job5, job6]]    #
        gevent.joinall(jobs, timeout=wait_params.get('timeout'))
        return True

    def activate_orderbook(self, mirror, start=3, end=6, k=1):
        '''
        params
            maker: DataFrame(columns=['price', 'quantity', 'side', 'id'])
            mirror: DataFrame([columns=['price', 'quantity', 'side'])
        '''
        long = mirror[mirror.side==Action.long].sort_values('price', ascending=False)
        short = mirror[mirror.side==Action.short].sort_values('price', ascending=True)

        long_top, short_top = long[long.price.pct_change()!=0], short[short.price.pct_change()!=0]
        long_top, short_top = long_top.iloc[start:end], short_top.iloc[start:end]
        long_top, short_top = long_top.sample(k), short_top.sample(k)
        long_top.quantity = mirror.quantity.quantile(0.1) * 1+random.random()
        long_top['type'] = 'send'
        short_top.quantity = mirror.quantity.quantile(0.1) * 1+random.random()
        short_top['type'] = 'send'
        long_action = long_top.to_dict('records')
        short_action = short_top.to_dict('records')
        
        def job1():
            for action in long_action:
                xt_api.send(self.symbol, action)
            if self.verbose:
                print('activate_orderbook', long_action)
        def job2():
            for action in short_action:
                xt_api.send(self.symbol, action)
            if self.verbose:
                print('activate_orderbook', short_action)
        jobs = [gevent.spawn(job) for job in [job1, job2]]
        gevent.joinall(jobs, timeout=wait_params.get('timeout'))
        return True

    def price_overflow(self, maker, side, method='batch'):
        o_action = self.find_overflow(maker, side=side)
        self.exec_outside_action(o_action)
        return True

    # metrics/optimal/control
    def risk_metrics(self):
        # balance
        unit = self.symbol.split('_')[1] if self.cfg.values('transcation_type') == 'um' else self.symbol.split('_')[0]
        wallet_balance = xt_api.balance(method='api', unit=unit, kind='walletBalance')
        init_fund = self.cfg.values('init_fund')
        total_pnl = round(wallet_balance-init_fund, 2)
        #
        available_balance = xt_api.balance(method='api', unit=unit, kind='availableBalance')
        balance_ratio = round((wallet_balance-available_balance)/wallet_balance, 2)
        # position
        position = xt_api.position(symbol=self.symbol, method='api')
        position = pd.DataFrame(position)
        position = position.astype({'realizedProfit': float, 'openOrderMarginFrozen': float})
        pos_pnl = position.realizedProfit.sum().round(2)
        # 
        long_position = position[position.positionSide=='LONG']
        short_position = position[position.positionSide=='SHORT']
        long_pos_ratio = (long_position.openOrderMarginFrozen/wallet_balance).round(2).values[0]
        short_pos_ratio = (short_position.openOrderMarginFrozen/wallet_balance).round(2).values[0]
        #
        # if any([long_pos_ratio>0.1, short_pos_ratio>0.1, balance_ratio>0.1]):
        content = f'市场: {self.symbol}\n持仓占比: {long_pos_ratio},{short_pos_ratio}\n使用资金占比: {balance_ratio}\n持仓盈亏: {pos_pnl}\n总盈亏:{total_pnl}'
        send_worktile_msg(content)
        print('risk metrics', content)
        return True

    def auto_optimize(self):
        time.sleep(1)
        if not int(time.time()) % wait_params.get('autopt_time'):
            # add merge position autopt
            if self.cfg.values('merge_autopt'):
                xt_api.merge_position(self.symbol)
            # TODO: add hedge autopt
            if self.cfg.values('hedge_autopt'):
                ...
            # TODO: add params autopt
            if self.cfg.values('params_autopt'):
                ...
        return True

    def controller(self):
        # TODO: test
        # balance_loss_control
        if all([not int(time.time())%wait_params.get('control_time'), \
            self.cfg.values('balance_loss_control')]):
            status = self.optimal.balance_loss()
            if status is False:
                redisdb.standalone_operator('hget', self.symbol+redis_optimal_state, ('stop_process', True))
                content = f'Stop liquidity in the {self.symbol} market forever'
                send_worktile_msg(content)
                print(content)
                self.stop_all()
                sys.exit(-1)
        # episode_loss_control
        if all([not int(time.time())%wait_params.get('control_time'), \
            self.cfg.values('episode_loss_control')]):
            diff_time = self.optimal.episode_loss()
            if diff_time > 0:
                redisdb.standalone_operator('hget', self.symbol+redis_optimal_state, ('temp_stop_liquid', diff_time))
                content = f'Temporarily stop {self.symbol} market liquidity {diff_time}'
                send_worktile_msg(content)
                print(content)
                # ref: send
        # keep_liquid_control
        #   ref: construct_quantity
        ...
        return True


class OrderBook(OrderBookMixins):

    def __init__(self, configure, event, engine, back) -> None:
        self.back = back
        super().__init__(configure, event, engine)
        self.init_xtapi(event, engine)
        self.init_optimal_state()

    @loop
    @exception
    def start_user_long_price_sync(self):
        time.sleep(wait_params.get('user_sync_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        mirror = self.mirror_data(observation='user')
        t2 = time.time()
        self.user_long_price_sync(maker, mirror)
        t3 = time.time()
        print('start_user_long_price_sync timing', t1-t0, t2-t1, t3-t2)
        print('-'*50)
        return True
    
    @loop
    @exception
    def start_mt_long_price_sync(self):
        time.sleep(wait_params.get('mt_sync_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        mirror = self.mirror_data(observation='mt')
        t2 = time.time()
        self.mt_long_price_sync(maker, mirror)
        t3 = time.time()
        print('start_mt_long_price_sync timing', t1-t0, t2-t1, t3-t2)
        print('-'*50)
        return True

    @loop
    @exception
    def start_user_short_price_sync(self):
        time.sleep(wait_params.get('user_sync_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        mirror = self.mirror_data(observation='user')
        t2 = time.time()
        self.user_short_price_sync(maker, mirror)
        t3 = time.time()
        print('start_user_short_price_sync timing', t1-t0, t2-t1, t3-t2)
        print('-'*50)
        return True
    
    @loop
    @exception
    def start_mt_short_price_sync(self):
        time.sleep(wait_params.get('mt_sync_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        mirror = self.mirror_data(observation='mt')
        t2 = time.time()
        self.mt_short_price_sync(maker, mirror)
        t3 = time.time()
        print('start_mt_short_price_sync timing', t1-t0, t2-t1, t3-t2)
        print('-'*50)
        return True

    @loop
    @exception
    def start_user_sub_density(self):
        time.sleep(wait_params.get('user_density_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        mirror = self.mirror_data(observation='user')
        t2 = time.time()
        self.sub_density(maker, mirror, side='long', observation='user')
        self.sub_density(maker, mirror, side='short', observation='user')
        t3 = time.time()
        print('start_user_sub_density timing', t1-t0, t2-t1, t3-t2)
        time.sleep(0.5)
        return True
    
    @loop
    @exception
    def start_mt_sub_density(self):
        time.sleep(wait_params.get('mt_density_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        mirror = self.mirror_data(observation='mt')
        t2 = time.time()
        self.sub_density(maker, mirror, side='long', observation='mt')
        self.sub_density(maker, mirror, side='short', observation='mt')
        t3 = time.time()
        print('start_mt_sub_density timing', t1-t0, t2-t1, t3-t2)
        time.sleep(0.5)
        return True

    @loop
    @exception
    def start_trades(self):
        time.sleep(0.1)
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        ...
        t1 = time.time()
        state = self.trade_sync_double()
        if not state:
            time.sleep(self.cfg.values('trade_freq'))
        t2 = time.time()
        print('start_trades timing', t1-t0, t2-t1)
        return True

    @loop
    @exception
    def start_quantity_sync(self):
        time.sleep(wait_params.get('quantity_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        mirror = self.mirror_data(observation='mt')
        t2 = time.time()
        self.quantity_sync(maker, mirror)
        t3 = time.time()
        print('start_quantity_sync timing', t1-t0, t2-t1, t3-t2)
        return True

    @loop
    @exception
    def start_activate_orderbook(self):
        time.sleep(wait_params.get('activate_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        mirror = self.mirror_data(observation='user')
        t1 = time.time()
        ...
        t2 = time.time()
        self.activate_orderbook(mirror)
        t3 = time.time()
        print('start_activate_orderbook timing', t1-t0, t2-t1, t3-t2)
        return True
    
    @loop
    @exception
    def start_aux_orderbook(self):
        time.sleep(wait_params.get('aux_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        ...
        t2 = time.time()
        self.aux_orderbook(maker)
        t3 = time.time()
        print('start_activate_orderbook timing', t1-t0, t2-t1, t3-t2)
        return True

    @loop
    @exception
    def start_long_overflow(self):
        time.sleep(wait_params.get('overflow_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        ...
        t2 = time.time()
        self.price_overflow(maker, side='long')
        t3 = time.time()
        print('start_long_overflow timing', t1-t0, t2-t1, t3-t2)
        time.sleep(0.5)
        return True
    
    @loop
    @exception
    def start_short_overflow(self):
        time.sleep(wait_params.get('overflow_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        ...
        t2 = time.time()
        self.price_overflow(maker, side='short')
        t3 = time.time()
        print('start_short_overflow timing', t1-t0, t2-t1, t3-t2)
        time.sleep(0.5)
        return True

    @loop
    @exception
    def start_long_intersection(self):
        time.sleep(wait_params.get('user_sync_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        ...
        t2 = time.time()
        action = self.find_outside_range(maker, 'long', is_is=True)
        self.exec_outside_action(action)
        t3 = time.time()
        print('start_long_intersection timing', t1-t0, t2-t1, t3-t2)
        time.sleep(0.5)
        return True
    
    @loop
    @exception
    def start_short_intersection(self):
        time.sleep(wait_params.get('user_sync_time'))
        if self.back.kill_now:
            sys.exit(0)
        t0 = time.time()
        maker = xt_api.open_orders(symbol=self.symbol)
        t1 = time.time()
        ...
        t2 = time.time()
        action = self.find_outside_range(maker, 'short', is_is=True)
        self.exec_outside_action(action)
        t3 = time.time()
        print('start_short_intersection timing', t1-t0, t2-t1, t3-t2)
        time.sleep(0.5)
        return True
    
    @loop
    @exception
    def start_risk_metrics(self):
        time.sleep(wait_params.get('risk_metrics_time'))
        if self.back.kill_now:
            sys.exit(0)
        self.risk_metrics()
        return True
    
    @loop
    @exception
    def start_optimize(self):
        if self.back.kill_now:
            sys.exit(0)
        self.auto_optimize()
        return True
    
    @loop
    @exception
    def start_timer(self):
        if self.back.kill_now:
            sys.exit(0)
        self.timer()
        return True

    @loop
    @exception
    def start_controller(self):
        if self.back.kill_now:
            sys.exit(0)
        self.controller()
        return True
    